package com.chukun.flink.table.api.sql;

import com.chukun.flink.table.bean.ClickBean;
import com.chukun.flink.table.source.PrepareData;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author chukun
 * @version 1.0.0
 * @description sql 使用事件时间窗口操作
 * @createTime 2022年06月03日 00:13:00
 */
public class GroupWindowEventTimeTemplate {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStreamSource<ClickBean> dataStream = env.fromCollection(PrepareData.getClicksData());

        SingleOutputStreamOperator<ClickBean> timedDataStream = dataStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.forGenerator((ctx) -> new WatermarkGenerator<ClickBean>() {
                    @Override
                    public void onEvent(ClickBean clickBean, long l, WatermarkOutput watermarkOutput) {
                    }

                    @Override
                    public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
                        watermarkOutput.emitWatermark(new Watermark(System.currentTimeMillis()));
                    }
                }).withTimestampAssigner((clickBean, timestamp) -> clickBean.getTime().getTime())
        );

        // 添加事件时间，额外的字段，这个字段必须在表模式的最后定义，该字段在提取时间戳和分配水印时分配时间戳
        // VisitTime为定义的逻辑字段名称，.rowtime()为定义的逻辑字段的后缀，表明该字段的属性是事件时间
        // Table inputTable = tableEnv.fromDataStream(timedDataStream,$("id"), $("user"), $("time"),$("url"),$("VisitTime").rowtime());

        // 添加事件时间，time字段用作提取时间戳，不在需要额外的字段提取事件时间
        // VisitTime为定义的逻辑字段名称，.rowtime()为定义的逻辑字段的后缀，表明该字段的属性是事件时间
        Table inputTable = tableEnv.fromDataStream(timedDataStream,$("id"), $("user"), $("VisitTime").rowtime(),$("url"));

        // tableEnv.registerDataStream("Clicks", timedDataStream, "id,user,time,url,VisitTime.rowtime");

//        tableEnv.createTemporaryView("Clicks", timedDataStream, Schema.newBuilder()
//                .column("id", DataTypes.INT())
//                .column("user", DataTypes.STRING())
//                .column("url", DataTypes.STRING())
//                .column("time", DataTypes.TIMESTAMP().bridgedTo(Timestamp.class))
//                .build());

        tableEnv.createTemporaryView("Clicks", inputTable);

        String sql = "select user as name, count(url), " +
                "tumble_start(VisitTime, interval '1' hour), " + // 窗口包含下界的开始时间
                "tumble_rowtime(VisitTime, interval '1' hour)," + // 窗口包含上界的结束时间
                "tumble_end(VisitTime, interval '1' hour) " +  // 窗口不包含上界的结束时间
                "from Clicks " +
                "group by tumble(VisitTime,interval '1' hour), user";

        Table table = tableEnv.sqlQuery(sql);

        DataStream<Row> rowDataStream = tableEnv.toChangelogStream(table);

        rowDataStream.print("窗口时间分组");

        env.execute("GroupWindowTemplate");

    }
}
